package mapreduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import app.App;
import app.Category;
import app.Service;
import util.hadoop.HDFSUtil;

public class ProcessApp {
	
	
	
	
	
/*	
	Flow no.,Flow[Src~Dst~Pro~Src Port~Dst Port],flow Length(s),Flow Throughput(kb/s),Bytes transferred per flow(bytes),Packets transferred per flow(个),Activity period
	1,[10.0.10.123~132.44.78.26~LLC~~],192.44871799999999,0.0011924848857657758,235.0,2,0~19
	2,[10.0.10.123~146.141.148.59~ARP~~],300.720508,0.0032863779612928826,1012.0,11,2~7~14~14~20~26~26~31~32~32~32
	3,[10.0.10.123~170.82.154.206~ARP~~],0.0023780000000215296,113.34367115120259,276.0,3,35~35~35*/
	
	public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			String[] flowInfo = value.toString().split(",");
			String[] flow = flowInfo[1].split("~");
			String src = flow[0];
			String dst = flow[1];
			context.write(new Text(src), new Text(dst));
		}

		@Override
		protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			context.nextKeyValue();
		}
		
	}
	
	public static class ReduceClass extends Reducer<Text, Text, NullWritable, Text> {
		
		App app;
		Service service;
		Category category;
		int ipNo = 0;
		
		
		@Override
		protected void setup(Reducer<Text, Text, NullWritable, Text>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			app = new App();
			service = new Service(app.getApp());
			category = new Category(app.getApp());
			String[] head = category.getHead();
			String out = "ipNo.,Src Ip,Service";
			for (String string : head) {
				out += "," + string;
			}
			out += "\r";
			context.write(NullWritable.get(), new Text(out));
		}

		@Override
		protected void reduce(Text arg0, Iterable<Text> arg1, Reducer<Text, Text, NullWritable, Text>.Context arg2)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			ipNo++;
			String src = arg0.toString().replace("[", "");
			List<String> domainList = new ArrayList<String>();
			for (Text text : arg1) {
				domainList.add(text.toString());
			}
			
			String serviceCode = service.getServiceCode(domainList);
			String code = category.getCategoryCode(domainList);
			String out = ipNo + "," + src + "," + serviceCode + "," + code;
			
			arg2.write(NullWritable.get(), new Text(out));
		}
		
		
		
		
	}
	
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		String master = args[0];
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", "hdfs://" + master + ":9000");
		
		HDFSUtil hdfsUtil = new HDFSUtil(conf);
		
		Job job = Job.getInstance(conf, "process");
		job.setJarByClass(ProcessApp.class);
		
		job.setMapperClass(MapClass.class);
		job.setReducerClass(ReduceClass.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		Path in = new Path("hdfs://" + master + ":9000/dataoutput/type3/type3A/");
		Path out = new Path("hdfs://" + master + ":9000/dataoutput/app/");
		hdfsUtil.delete(out);
		FileInputFormat.addInputPath(job, in);
		FileOutputFormat.setOutputPath(job, out);
		
		System.out.println(job.waitForCompletion(true) ? 0 : 1);
		
		Path out1 = new Path("hdfs://" + master + ":9000/dataoutput/app/_SUCCESS");
		hdfsUtil.delete(out1);

		
	}

}
